package uploader

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"sync"
	"time"

	"ug-uploads/config"
	"ug-uploads/logx"

	tos "github.com/volcengine/ve-tos-golang-sdk/v2/tos"
	"github.com/volcengine/ve-tos-golang-sdk/v2/tos/enum"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

type rateLimit struct {
	rate     int64
	capacity int64

	currentAmount int64
	sync.Mutex
	lastConsumeTime time.Time
}

func NewDefaultRateLimit(rate int64, capacity int64) tos.RateLimiter {
	return &rateLimit{
		rate:            rate,
		capacity:        capacity,
		lastConsumeTime: time.Now(),
		currentAmount:   capacity,
		Mutex:           sync.Mutex{},
	}
}

func (d *rateLimit) Acquire(want int64) (ok bool, timeToWait time.Duration) {
	d.Lock()
	defer d.Unlock()
	if want > d.capacity {
		want = d.capacity
	}
	increment := int64(time.Now().Sub(d.lastConsumeTime).Seconds() * float64(d.rate))
	if increment+d.currentAmount > d.capacity {
		d.currentAmount = d.capacity
	} else {
		d.currentAmount += increment
	}
	if want > d.currentAmount {
		timeToWaitSec := float64(want-d.currentAmount) / float64(d.rate)
		return false, time.Duration(timeToWaitSec * float64(time.Second))
	}
	d.lastConsumeTime = time.Now()
	d.currentAmount -= want
	return true, 0
}

type listener struct {
}

func (l *listener) DataTransferStatusChange(event *tos.DataTransferStatus) {
	switch event.Type {
	case enum.DataTransferStarted:
		fmt.Println("当前分片准备中...")
	case enum.DataTransferRW:
		// Chunk 模式下 TotalBytes 值为 -1
		if event.TotalBytes != -1 {
			// fmt.Printf("分片上传进度:%d,ConsumerBytes/TotalBytes: %d/%d,%d%%\n", event.RWOnceBytes, event.ConsumedBytes, event.TotalBytes, event.ConsumedBytes*100/event.TotalBytes)
			fmt.Printf("当前文件块流大小: %d kb, 已上传/总大小: %d kb / %d kb, 当前分片进度: %d%%\n", event.RWOnceBytes/1024, event.ConsumedBytes/1024, event.TotalBytes/1024, event.ConsumedBytes*100/event.TotalBytes)
		} else {
			// fmt.Printf("普通上传进度:%d,ConsumerBytes:%d\n", event.RWOnceBytes, event.ConsumedBytes)
			// fmt.Printf("普通上传进度:%d\n", event.ConsumedBytes*100/event.TotalBytes)
		}
	case enum.DataTransferSucceed:
		// fmt.Printf("分片上传成功, ConsumerBytes/TotalBytes: %d/%d, 进度: %d%%\n", event.ConsumedBytes, event.TotalBytes, event.ConsumedBytes*100/event.TotalBytes)
		fmt.Println("当前分片上传成功!")
	case enum.DataTransferFailed:
		fmt.Printf("上传出错...\n")
	}
}

func UploadToTOS(ctx context.Context, advertiserID string, objectName, fileURL, posterName, posterURL string) error {
	logx.Logger.Info("开始上传素材", zap.String("advertiser_id", advertiserID))

	client, err := tos.NewClientV2(
		config.Cfg.TOSEndpoint,
		tos.WithRegion(config.Cfg.TOSRegion),
		tos.WithCredentials(tos.NewStaticCredentials(config.Cfg.AccessKey, config.Cfg.SecretKey)),
	)

	if err != nil {
		return fmt.Errorf("创建 TOS 客户端失败: %w", err)
	}

	downloadCtx, cancelDownload := context.WithTimeout(ctx, 10*time.Minute)
	defer cancelDownload()
	g, _ := errgroup.WithContext(ctx)
	var videoResp, posterResp *http.Response

	g.Go(func() error {
		resp, err := downloadFile(downloadCtx, fileURL)

		if err != nil {
			return fmt.Errorf("下载视频失败: %w", err)
		}

		videoResp = resp
		return nil
	})

	g.Go(func() error {
		resp, err := downloadFile(downloadCtx, posterURL)

		if err != nil {
			return fmt.Errorf("下载封面失败: %w", err)
		}

		posterResp = resp
		return nil
	})

	if err := g.Wait(); err != nil {
		return err
	}

	defer videoResp.Body.Close()
	defer posterResp.Body.Close()

	if videoResp.ContentLength <= 0 {
		return fmt.Errorf("视频文件大小为 0")
	}

	// fmt.Println("")
	// fmt.Println("正在执行广告主 id 为:", advertiserID, "的素材上传==================================================>")
	// fmt.Println("")
	// fmt.Println("")

	uploadCtx, cancelUpload := context.WithTimeout(ctx, 30*time.Minute)
	defer cancelUpload()
	uploadGroup, _ := errgroup.WithContext(uploadCtx)

	uploadGroup.Go(func() error {
		return uploadToTOSObject(uploadCtx, client, posterName, posterResp.Body, "封面")
	})

	uploadGroup.Go(func() error {
		//if videoResp.ContentLength >= 5*1024*1024*1024 { // 5GB
		return multipartUpload(uploadCtx, client, objectName, videoResp.Body)
		//}
		//return normalUpload(ctx, client, cfg, objectName, videoResp.Body, logger)
	})

	if err := uploadGroup.Wait(); err != nil {
		return fmt.Errorf("素材上传失败: %w", err)
	}

	logx.Logger.Info("素材上传完成", zap.String("advertiser_id", advertiserID), zap.String("object_name", objectName))
	return nil
}

func downloadFile(ctx context.Context, url string) (*http.Response, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)

	if err != nil {
		return nil, fmt.Errorf("创建请求失败: %w", err)
	}

	resp, err := http.DefaultClient.Do(req)

	if err != nil {
		return nil, fmt.Errorf("请求失败: %w", err)
	}

	if resp.StatusCode != http.StatusOK {
		resp.Body.Close()
		return nil, fmt.Errorf("下载失败, HTTP 状态码: %s", resp.Status)
	}

	return resp, nil
}

func uploadToTOSObject(ctx context.Context, client *tos.ClientV2, key string, body io.Reader, label string) error {
	rateLimit1M := 1024 * 1024 * 5

	put := &tos.PutObjectV2Input{
		PutObjectBasicInput: tos.PutObjectBasicInput{
			Bucket:      config.Cfg.BucketName,
			Key:         key,
			RateLimiter: NewDefaultRateLimit(int64(rateLimit1M), int64(rateLimit1M)),
		},
		Content: body,
	}

	out, err := client.PutObjectV2(ctx, put)

	if err != nil {
		return fmt.Errorf("%s上传失败: %w", label, err)
	}

	logx.Logger.Info(fmt.Sprintf("%s上传成功", label), zap.String("key", key), zap.String("request_id", out.RequestID))
	return nil
}

func multipartUpload(ctx context.Context, client *tos.ClientV2, key string, reader io.Reader) error {
	rateLimit1M := 1024 * 1024 * 5

	createOut, err := client.CreateMultipartUploadV2(ctx, &tos.CreateMultipartUploadV2Input{
		Bucket: config.Cfg.BucketName,
		Key:    key,
	})

	if err != nil {
		return fmt.Errorf("创建分片上传失败: %w", err)
	}

	uploadID := createOut.UploadID
	const partSize = 20 * 1024 * 1024
	buf := make([]byte, partSize)
	var parts []tos.UploadedPartV2
	partNum := 1

	for {
		n, readErr := io.ReadFull(reader, buf)

		if n == 0 && readErr == io.EOF {
			break
		}

		if readErr != nil && readErr != io.ErrUnexpectedEOF && readErr != io.EOF {
			return readErr
		}

		partReader := bytes.NewReader(buf[:n])

		out, uploadErr := client.UploadPartV2(ctx, &tos.UploadPartV2Input{
			UploadPartBasicInput: tos.UploadPartBasicInput{
				Bucket:               config.Cfg.BucketName,
				Key:                  key,
				UploadID:             uploadID,
				PartNumber:           partNum,
				DataTransferListener: &listener{},
				RateLimiter:          NewDefaultRateLimit(int64(rateLimit1M), int64(rateLimit1M)),
			},
			Content:       partReader,
			ContentLength: int64(n),
		})

		if uploadErr != nil {
			return fmt.Errorf("分片上传失败: %w", uploadErr)
		}

		logx.Logger.Info("分片上传成功", zap.Int("part", partNum))
		parts = append(parts, tos.UploadedPartV2{PartNumber: partNum, ETag: out.ETag})
		partNum++

		if readErr == io.EOF || readErr == io.ErrUnexpectedEOF {
			break
		}
	}

	_, err = client.CompleteMultipartUploadV2(ctx, &tos.CompleteMultipartUploadV2Input{
		Bucket:   config.Cfg.BucketName,
		Key:      key,
		UploadID: uploadID,
		Parts:    parts,
	})

	if err != nil {
		return fmt.Errorf("完成分片上传失败: %w", err)
	}

	logx.Logger.Info("素材上传完成", zap.String("objectName", key))
	return nil
}

// 普通的上传
// func normalUpload(ctx context.Context, client *tos.ClientV2, cfg *config.Config, key string, body io.Reader, logger *zap.Logger) error {
// 	put := &tos.PutObjectV2Input{
// 		PutObjectBasicInput: tos.PutObjectBasicInput{
// 			Bucket:               cfg.BucketName,
// 			Key:                  key,
// 			DataTransferListener: &listener{},
// 		},
// 		Content: body,
// 	}
// 	out, err := client.PutObjectV2(ctx, put)
// 	if err != nil {
// 		return fmt.Errorf("普通上传失败: %w", err)
// 	}
// 	logger.Info("普通上传成功", zap.String("key", key), zap.String("request_id", out.RequestID))
// 	return nil
// }
